package com.managertrade.job;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.managertrade.common.SystemConfigEnum;
import com.managertrade.common.ThreadPoolEnum;
import com.managertrade.config.MultiPeriodCache;
import com.managertrade.config.OkxTraderCache;
import com.managertrade.dao.sqlite.entity.KlineWithIndicatorsEntity;
import com.managertrade.dto.*;
import com.managertrade.dto.vo.KlineWithIndicators;
import com.managertrade.dto.vo.OkxLeadTraderResponseVO;
import com.managertrade.dao.mysql.entity.OkxOrder;
import com.managertrade.dao.mysql.entity.SystemConfigEntity;
import com.managertrade.dao.mysql.entity.UserCopyTrade;
import com.managertrade.dao.mysql.entity.UserFollowTrader;
import com.managertrade.dao.mysql.repository.OkxOrderRepository;
import com.managertrade.dao.mysql.repository.SystemConfigRepository;
import com.managertrade.dao.mysql.repository.UserCopyTradeRepository;
import com.managertrade.dao.mysql.repository.UserFollowTraderRepository;
import com.managertrade.service.impl.KlineService;
import com.managertrade.service.sqllite.service.KlineWithIndicatorsService;
import com.managertrade.service.impl.LockService;
import com.managertrade.util.OkxPerpClient;
import com.managertrade.util.QQEmailSenderUtil;
import com.managertrade.util.strategy.TradeSignalUtil;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.tuple.Pair;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.*;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StopWatch;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.text.SimpleDateFormat;
import java.time.*;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

@EnableScheduling
@Component
@Slf4j
public class EmaMacdTrendHuntingStrategy {
    @Autowired
    private KlineService marketService;
    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    KlineService klineService;
    @Resource
    OkxTraderCache okxTraderCache;
    @Autowired
    MultiPeriodCache multiPeriodCache;
    @Autowired
    private UserFollowTraderRepository userFollowTraderRepository;

    @Autowired
    private OkxOrderRepository okxOrderRepository;

    @Autowired
    private UserCopyTradeRepository userCopyTradeRepository;
    @Autowired
    private SystemConfigRepository systemConfigRepository;
    @Autowired
    KlineWithIndicatorsService klineWithIndicatorsService;

    @Autowired
    private LockService lockService;


    private static final long HOUR_MS = 60 * 60 * 1000L;
    private static final int WINDOW_COUNT = 300; // 推算300个周期
    private static final String CACHE_KEY = "kline_time_window";
    private static final ConcurrentHashMap<String, ReentrantLock> locks = new ConcurrentHashMap<>();
    public static ReentrantLock getLock(String name) {
        log.info("开始获取锁name{}",name);
        // 如果锁不存在就创建一个新的锁
        return locks.computeIfAbsent(name, k -> new ReentrantLock());
    }
   //@Scheduled(cron = "0 * * * * ?")
    public void executeProduceExcelTask30m() throws Exception {
        Pair<Long, Long> longLongPair = generateAlignedRandomTimeRange(80);
        Long start = longLongPair.getLeft();
        Long end = longLongPair.getRight();
        log.info("当前回测时间开始"+TradeSignalUtil.formatTimestamp(start)+"结束时间："+TradeSignalUtil.formatTimestamp(end));
        //获取1小时数据
        List<KlineWithIndicators> klineWithIndicatorsOneHour = marketService.getKlineWithIndicators("BTCUSDT", "1h", null, "binance",start,end);
        //获取4小时数据
        List<KlineWithIndicators> klineWithIndicatorsFourHour = marketService.getKlineWithIndicators("BTCUSDT", "4h", null, "binance",start,end);
        //判断是否找到符合条件的数据
        Map<Boolean, String> tradeSignal = TradeSignalUtil.isTradeSignal(klineWithIndicatorsOneHour, klineWithIndicatorsFourHour);
        List<String> allMatchingSignals = TradeSignalUtil.findAllMatchingSignals(klineWithIndicatorsOneHour, klineWithIndicatorsFourHour);
        //发送邮件
        String result = tradeSignal.get(Boolean.TRUE);
        if(!StringUtils.isEmpty(result)){
            StringBuilder str =  new StringBuilder();
            allMatchingSignals.stream().map(k->str.append(k)).collect(Collectors.toList());
            QQEmailSenderUtil.sendEmail("macd+ema7-21策略信号提醒",str.toString());
        }else{
            log.info("当前回测时间"+TradeSignalUtil.formatTimestamp(start)+"未找到符合条件的数据");
        }
    }


    //@Scheduled(cron = "*/1 * * * * ?")
    public void executeTrainPythonTask30m() throws Exception {
        Pair<Long, Long> longLongPair = generateAlignedRandomTimeRange(80);
        Long start = longLongPair.getLeft();
        Long end = longLongPair.getRight();

        log.info("模型训练当前回测时间开始：" + TradeSignalUtil.formatTimestamp(start) + "，结束时间：" + TradeSignalUtil.formatTimestamp(end));

        List<KlineWithIndicators> klineList = marketService.getKlineWithIndicators("BTCUSDT", "1h", 800, "binance", start, end);

        if (klineList == null || klineList.size() < 20) {
            log.warn("K线数据不足，跳过训练！");
            return;
        }

        String jsonData = JSONArray.toJSONString(klineList, SerializerFeature.WriteMapNullValue);

        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);

        HttpEntity<String> requestEntity = new HttpEntity<>(jsonData, headers);

        String flaskUrl = "http://127.0.0.1:5000/train";

        try {
            ResponseEntity<TrainResponseDTO> response = restTemplate.exchange(
                    flaskUrl, HttpMethod.POST, requestEntity, TrainResponseDTO.class);

            TrainResponseDTO trainResponse = response.getBody();

            if (trainResponse != null) {
                log.info("训练轮次: " + trainResponse.getRound());
                log.info("状态: " + trainResponse.getStatus());
                log.info("准确率: " + trainResponse.getAccuracy());
                log.info("模型权重: " + trainResponse.getCoef());
                if (trainResponse.getFutureDetails() != null) {
                    log.info("未来5根K线价格: " + trainResponse.getFutureDetails().getFuturePrices());
                    log.info("未来5根K线涨跌趋势: " + trainResponse.getFutureDetails().getFutureTrends());
                }
            } else {
                log.warn("Python训练接口返回空响应");
            }
        } catch (Exception e) {
            log.error("调用Python训练服务失败", e);
        }
    }

    public static Pair<Long, Long> generateAlignedRandomTimeRange(int count4h) {
        long fourHourMillis = 4 * 60 * 60 * 1000;
        long totalMillis = count4h * fourHourMillis;
        long minTime = Instant.parse("2017-08-02T00:00:00Z").toEpochMilli();
        long maxStart = System.currentTimeMillis() - totalMillis;

        long start = minTime + (long)(Math.random() * (maxStart - minTime));
        long end = start + totalMillis;

        return Pair.of(start, end);
    }

    /**
     * 获取历史的数据从2017-08-01 日到当前日期  1小时  4小时  6小时  12小时  1天  k线数据 存入当前项目目前，存入的id数据就放入缓存id,每次都校验
     * 获取当前的excel 依据唯一id   存在就更新和新增数据  不存在就新增该文件和目录
     */
    @Scheduled(cron = "0 * * * * ?")
    public void produceExcel() throws Exception {
        StopWatch stopWatch =  new StopWatch();
        stopWatch.start();
        List<String> peridList = Arrays.asList("1m","3m","5m","1h");
        List<String> symbolList = Arrays.asList("ASTERUSDT");
        List<Future<?>> futures = new ArrayList<>();
        for(String symbol:symbolList){
            for(String period: peridList){
                try {
                    Future<?> future =  ThreadPoolEnum.SINGLE_THREAD_POOL.getExecutor().submit(()->{
                        TimeRange timeRange = getKlineWithIndicatorsEntity(symbol,period);
                        log.info(" symbol{}：period{}获取生成表格时间"+timeRange.toString(),symbol,period);
                        //获取2017-08-01 后的每个整点小时时间戳  开始时间和结束时间  中间是300个小时的间隔，每次都把最新的获取过结束时间放入缓存的，如果获取过，就依次变缓存改变到下一个要获取的时间戳
                        List<KlineWithIndicators> klineList = null;
                        try {
                            klineList = marketService.getKlineWithIndicators(symbol, period, null, "binance", timeRange.start, timeRange.end);
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
                        log.info("symbol{}：period{}获取 数据条数{}",symbol,period,klineList.size());
                        //更新到sqlite
                        klineWithIndicatorsService.saveIfNotExists(klineList);
                    });
                    futures.add(future);
                }catch (Exception e){
                    TimeRange timeRange = afterTimeRange(symbol, period);
                    log.info("回退上一个周期计算symbol{}：period{}  回退信息{} 错误{}",symbol,symbol,timeRange.toString(),e.getStackTrace());
                }


            }
        }

        // 等待所有任务完成
        for (Future<?> future : futures) {
            future.get();  // 阻塞，直到任务执行完
        }
        stopWatch.stop();
        log.info("一次耗时ms：{}",stopWatch.getTotalTimeMillis());
    }



    /**
     * 获取指定周期的下一个 300 小时的时间窗口
     * 每次返回 TimeRange，同时把最新时间戳放入缓存
     */
    public TimeRange nextTimeRange(String symbol,String period) {
        long step = getPeriodMillis(period);
        long startTs;

       // 先从表格或者缓存取最新时间
        Object excelValue = multiPeriodCache.get(symbol + "_" + period);
        if (excelValue != null) {
            startTs = (long) excelValue; // 从最新数据后一条开始
        } else {
            if(symbol.contains("BNB")){
                startTs = LocalDateTime.parse("2017-11-06T08:00:00")
                        .toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
            }else {
                startTs = LocalDateTime.parse("2017-08-14T00:00:00")
                        .toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
            }
        }

        long endTs = startTs + step*300; // 拉取一定区间
        long nowTs = System.currentTimeMillis();
        if(endTs > nowTs) endTs = nowTs; // 避免未来时间
        // 更新缓存，下次从最新表格时间开始
        multiPeriodCache.put(symbol + "_" + period, endTs);
        return new TimeRange(startTs, endTs);

    }

    /**
     * 获取最新币种的币种 周期 数据
     * @param period
     * @return
     */
    public TimeRange getKlineWithIndicatorsEntity(String symbol, String period){
        KlineWithIndicatorsEntity topBySymbolAndPeriodOrderByTimestampDesc = klineWithIndicatorsService.findTopBySymbolAndPeriodOrderByTimestampDesc(symbol, period);
        if(topBySymbolAndPeriodOrderByTimestampDesc == null){
            return nextTimeRange(symbol,period);
        }
        long step = getPeriodMillis(period);
        // 先从表格或者缓存取最新时间
        Object excelValue = topBySymbolAndPeriodOrderByTimestampDesc.getTimestamp();
        long  startTs = (long) excelValue;
        long endTs = startTs + step*300; // 拉取一定区间
        return new TimeRange(startTs, endTs);
    }

    // 根据周期返回毫秒数
    private long getPeriodMillis(String period) {
        switch (period) {
            case "1m": return 60_000L;
            case "3m": return 180_000L;
            case "5m": return 300_000L;
            case "1h": return 3600_000L;
            case "4h": return 4 * 3600_000L;
            case "6h": return 6 * 3600_000L;
            case "12h": return 12 * 3600_000L;
            case "1d": return 24 * 3600_000L;
            default: throw new IllegalArgumentException("Unsupported period: " + period);
        }
    }

    public TimeRange afterTimeRange(String symbol,String period) {
        long step = getPeriodMillis(period);
        long startTs;

       // 先从表格或者缓存取最新时间
        Object excelValue = multiPeriodCache.get(symbol + "_" + period);
        if (excelValue != null) {
            startTs = (long) excelValue - step; // 从最新数据后一条开始
        } else {
            startTs = LocalDateTime.parse("2017-08-14T00:00:00")
                    .toInstant(ZoneOffset.UTC).toEpochMilli();
        }

        long endTs = startTs + step*300; // 拉取一定区间
        long nowTs = System.currentTimeMillis();
        if(endTs > nowTs) endTs = nowTs; // 避免未来时间
        // 更新缓存，下次从最新表格时间开始
        multiPeriodCache.put(symbol + "_" + period, endTs);
        return new TimeRange(startTs, endTs);
    }
    // 根据周期返回毫秒数

    /**
     * 时间区间对象
     */
    public static class TimeRange {
        public final long start;
        public final long end;

        public TimeRange(long start, long end) {
            this.start = start;
            this.end = end;
        }

        public String toString() {
            DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            String s = LocalDateTime.ofInstant(Instant.ofEpochMilli(start), ZoneOffset.UTC).format(fmt);
            String e = LocalDateTime.ofInstant(Instant.ofEpochMilli(end), ZoneOffset.UTC).format(fmt);
            return String.format("[%s -> %s]", s, e);
        }
    }


    //@Scheduled(cron = "0 0/2 * * * ?")
    public void tradeUserModel() {
        ThreadPoolEnum.PREDICT_SINGLE_THREAD_POOL.getExecutor().submit(() -> {
            String lockName = "LOCK_COPY_TRADE_TASK";
            if (!lockService.tryLock(lockName)) {
                log.info("其他服务器正在执行任务，直接跳过");
                return;
            }

            StopWatch stopWatch = new StopWatch();
            stopWatch.start();
            log.info("开始获取交易 {}", new Date());

            boolean shouldRun = true; // 默认执行

            try {
                // 1. 获取关注的跟单员
                List<UserFollowTrader> allTraders = getActiveFollowTraders();
                if (CollectionUtils.isEmpty(allTraders)) {
                    log.info("没有关注的交易员，跳过执行");
                    shouldRun = false;
                }

                // 2. 获取交易员仓位信息
                Map<String, List<LeadTraderCurrentPositionsResponseDTO.Position>> dataTotal =
                        shouldRun ? getLeadTraderPositions(allTraders) : Collections.emptyMap();
                if (shouldRun && dataTotal.isEmpty()) {
                    log.info("没有交易员仓位信息，跳过执行");
                    shouldRun = false;
                }

                // 3. 获取高胜率交易员
                List<OkxLeadTraderResponseVO.Trader> topTraders =
                        shouldRun ? getTopTraders(allTraders, dataTotal) : Collections.emptyList();
                if (shouldRun && topTraders.isEmpty()) {
                    log.info("没有高胜率交易员，跳过执行");
                    shouldRun = false;
                }

                // 4. 获取自己当前仓位
                Map<String, OkxPositionResponseDTO.PositionData> myPositions =
                        shouldRun ? getMyPositions() : Collections.emptyMap();

                // 5. 获取交易员最新可操作仓位
                Map<String, LeadTraderCurrentPositionsResponseDTO.Position> traderPositionMap =
                        shouldRun ? getLatestTraderPositions(topTraders, dataTotal, myPositions) : Collections.emptyMap();

                if (shouldRun && traderPositionMap.isEmpty()) {
                    log.info("设置定时范围内的最新交易员仓位不存在");
                    shouldRun = false;
                }

                // 6. 获取币种信息
                Map<String, InstrumentResponseDTO.Instrument> instrumentsInfo =
                        shouldRun ? OkxPerpClient.getInstruments() : Collections.emptyMap();

                // 7. 遍历交易员下单
                if (shouldRun) {
                    for (UserFollowTrader trader : allTraders) {
                        handleTraderOrder(trader, traderPositionMap, myPositions, instrumentsInfo);
                    }
                }

            } catch (Exception e) {
                log.error("交易策略执行异常", e);
            } finally {
                lockService.releaseLock(lockName);
                stopWatch.stop();
                log.info("释放分布式锁: {}", lockName);
                log.info("执行一次交易策略耗时 {} ms", stopWatch.getTotalTimeMillis());
            }
        });
    }



    /** 获取活跃的跟单员 */
    private List<UserFollowTrader> getActiveFollowTraders() {
        List<UserFollowTrader> all = userFollowTraderRepository.findAll();
        long threeDaysAgo = System.currentTimeMillis() - 3L * 24 * 60 * 60 * 1000;
        return all.stream()
                .filter(trader -> trader.getUpdateOpentime() == null || trader.getUpdateOpentime() > threeDaysAgo)
                .filter(UserFollowTrader::getIsFollowing)
                .collect(Collectors.toList());
    }

    /** 获取交易员仓位 */
    private Map<String, List<LeadTraderCurrentPositionsResponseDTO.Position>> getLeadTraderPositions(List<UserFollowTrader> traders) {
        Map<String, List<LeadTraderCurrentPositionsResponseDTO.Position>> dataTotal = new ConcurrentHashMap<>();
        long nowStart = getCurrentMillisNY();
        long plus = 1000*60L*5;

        for (UserFollowTrader trader : traders) {
            try {
                LeadTraderCurrentPositionsResponseDTO positions = OkxPerpClient.getCurrentLeadPositions(trader.getLeadTraderId());
                if ("0".equals(positions.getCode()) && !CollectionUtils.isEmpty(positions.getData())) {
                    List<LeadTraderCurrentPositionsResponseDTO.Position> filtered = positions.getData().stream().filter(k->k.getOpenTime()!=null)
                            .filter(p -> !"".equals(p.getInstId()) && !StringUtils.isEmpty(getPositionDirection(p)))
                            .filter(p -> p.getOpenTime() > nowStart - plus)
                            .collect(Collectors.toList());
                    if (!filtered.isEmpty()) dataTotal.put(trader.getLeadTraderId(), positions.getData());
                }
            } catch (Exception ignored) {}
        }
        log.info("获取时间大于当天凌晨拥有仓位的跟单者数量{}", dataTotal.size());
        return dataTotal;
    }

    /** 获取高胜率交易员 */
    private List<OkxLeadTraderResponseVO.Trader> getTopTraders(List<UserFollowTrader> traders,
                                                               Map<String, List<LeadTraderCurrentPositionsResponseDTO.Position>> dataTotal) {
        List<String> traderIds = traders.stream().map(UserFollowTrader::getLeadTraderId)
                .filter(dataTotal::containsKey).collect(Collectors.toList());
        Map<String, OkxLeadTraderResponseVO.Trader> batchCache = okxTraderCache.getBatchCache(traderIds);
        if (MapUtils.isEmpty(batchCache)) return Collections.emptyList();

        return batchCache.values().stream()
                .filter(t -> !StringUtils.isEmpty(t.getWinRatio()))
                .sorted((t1, t2) -> Double.compare(Double.parseDouble(t2.getWinRatio()),
                        Double.parseDouble(t1.getWinRatio())))
                .limit(200)
                .collect(Collectors.toList());
    }

    /** 获取自己当前仓位 */
    private Map<String, OkxPositionResponseDTO.PositionData> getMyPositions() {
        Map<String, OkxPositionResponseDTO.PositionData> map = new HashMap<>();
        try {
            OkxPositionResponseDTO positions = OkxPerpClient.getAccountPositions();
            if (positions != null && "0".equals(positions.getCode()) && !CollectionUtils.isEmpty(positions.getData())) {
                map.putAll(positions.getData().stream()
                        .collect(Collectors.toMap(OkxPositionResponseDTO.PositionData::getInstId, p -> p)));
            }
        } catch (Exception e) {
            log.info("获取仓位信息有错误");
        }
        return map;
    }

    /** 获取交易员最新可操作仓位 */
    private Map<String, LeadTraderCurrentPositionsResponseDTO.Position> getLatestTraderPositions(
            List<OkxLeadTraderResponseVO.Trader> topTraders,
            Map<String, List<LeadTraderCurrentPositionsResponseDTO.Position>> dataTotal,
            Map<String, OkxPositionResponseDTO.PositionData> myPositions) {

        Map<String, LeadTraderCurrentPositionsResponseDTO.Position> map = new HashMap<>();
        long now = getCurrentUsEastMillis();
        long threeMinutesMillis = 30 * 60 * 1000L;

        for (OkxLeadTraderResponseVO.Trader trader : topTraders) {
            List<LeadTraderCurrentPositionsResponseDTO.Position> positions = dataTotal.get(trader.getUniqueCode());
            if (CollectionUtils.isEmpty(positions)) continue;

            positions.stream()
                    .filter(p -> "SWAP".equals(p.getInstType()))
                    .filter(p -> p.getOpenTime() <= now && now - p.getOpenTime() <= threeMinutesMillis)
                    .max(Comparator.comparingLong(LeadTraderCurrentPositionsResponseDTO.Position::getOpenTime))
                    .ifPresent(position -> map.put(trader.getUniqueCode(), position));
        }

        log.info("设置定时范围内的最新交易员仓位存在 数量{}", map.size());
        return map;
    }

    /** 处理单个交易员下单逻辑 */
    private void handleTraderOrder(UserFollowTrader trader,
                                   Map<String, LeadTraderCurrentPositionsResponseDTO.Position> traderPositionMap,
                                   Map<String, OkxPositionResponseDTO.PositionData> myPositions,
                                   Map<String, InstrumentResponseDTO.Instrument> instrumentsInfo) throws Exception {

        LeadTraderCurrentPositionsResponseDTO.Position position = traderPositionMap.get(trader.getLeadTraderId());
        if (position == null) return;

        String direction = getPositionDirection(position);

        // 检查自己仓位方向
        OkxPositionResponseDTO.PositionData myPos = myPositions.get(position.getInstId());
        if (myPos != null) {
            double posQty = Double.parseDouble(myPos.getPos());
            if ((posQty > 0 && "buy".equals(direction)) || (posQty < 0 && "sell".equals(direction))) {
                log.info("交易员：{} symbol:{} 持仓方向与预测方向一致，不需要下单", trader.getLeadTraderId(), position.getInstId());
                return;
            }
        }

        // 校验是否已经下过单
        if (!validateOrderExist(position.getOpenTime(),trader.getLeadTraderId(), position.getSubPosId(), position.getInstId())) {
            // 反向处理
            SystemConfigEntity config = systemConfigRepository.findByParamKey(SystemConfigEnum.BUSINESSSETTINGS_SIDE.name());
            if (config != null) direction = direction.equals("buy") ? "sell" : "buy";

            // 下单
            OkxPlaceOrderResultDTO result = OkxPerpClient.placeOrderFutureOkex(position.getInstId(), direction.toLowerCase(),
                    instrumentsInfo, new BigDecimal(0.1));
            saveOrderInfo(trader.getUserId(), result.getRequest(), result.getResponse().getData().get(0),
                    trader.getLeadTraderId(), position.getSubPosId(),position.getOpenTime());

            log.info("交易员 {} 下单成功, 仓位SubPosId {}, 下单ID {}",
                    trader.getLeadTraderId(), position.getSubPosId(), result.getResponse().getData().get(0).getOrdId());
        }
    }


    /**
     * 保存下单请求和返回数据
     *
     * @param request 下单请求 DTO
     * @param response 下单返回 DTO
     * @param unicode 用户或交易员唯一标识
     * @return 保存后的 OkxOrder 实体
     */
    public OkxOrder saveOrderInfo(Long userId,OkxPlaceOrderRequestDTO request, OkxOrderResponseDTO.OrderData response, String unicode,String subPosId,Long OpenTime) {
        OkxOrder okxOrder = new OkxOrder();
        // ==================== 请求相关 ====================
        okxOrder.setUserId(userId);
        okxOrder.setUnicode(unicode);
        okxOrder.setInstId(request.getInstId());
        okxOrder.setTdMode(request.getTdMode());
        okxOrder.setSide(request.getSide());
        okxOrder.setPosSide(request.getPosSide());
        okxOrder.setOrdType(request.getOrdType());
        okxOrder.setSz(!StringUtils.isEmpty(request.getSz())?new BigDecimal(request.getSz()):new BigDecimal(0));
        okxOrder.setReduceOnly(request.getReduceOnly() != null ? request.getReduceOnly() : false);
        okxOrder.setPx(!StringUtils.isEmpty(request.getPx())?new BigDecimal(request.getPx()):new BigDecimal(0));
        okxOrder.setCcy(request.getCcy() != null ? request.getCcy() : "USDT");

        //保存订单和跟单者关系
        UserCopyTrade trade  = new UserCopyTrade();
        trade.setUserId(userId);
        trade.setTraderId(unicode);
        // ==================== 响应相关 ====================
        if (response != null) {
            okxOrder.setOrdId(response.getOrdId());
            okxOrder.setClOrdId(response.getClOrdId());
            okxOrder.setSCode(response.getSCode());
            okxOrder.setSMsg(response.getSMsg());
            okxOrder.setTag(response.getTag());
            okxOrder.setTs(Long.valueOf(response.getTs()));
            trade.setOrdId(response.getOrdId());
            trade.setOpenTime(new Date(Long.valueOf(response.getTs())));
        }

        // ==================== 时间 ====================
        okxOrder.setCreateTime(LocalDateTime.now());
        okxOrder.setUpdateTime(LocalDateTime.now());
        trade.setSubPosId(subPosId);
        trade.setStatus("OPEN");
        trade.setPosSide(request.getSide());
        trade.setPos(!StringUtils.isEmpty(request.getSz())?new BigDecimal(request.getSz()):new BigDecimal(0));
        trade.setMgnMode(request.getTdMode());
        trade.setInstId(request.getInstId());
        trade.setLeadTraderOpenTime(new Date(OpenTime));
        userCopyTradeRepository.save(trade);
        // 保存
        return okxOrderRepository.save(okxOrder);
    }


    /**
     * 保存跟单盈利信息
     * @param leadTraderId
     * @param
     */
    @Transactional
    public void saveOrUpdateWatchTrader(Long userId,String leadTraderId,String instId,String orderId) throws Exception {
        //获取账单盈利
        BigDecimal pnl = OkxPerpClient.getOrderPnl(instId, orderId);
        log.info("当前跟单员{}账户持仓订单号：{}，币种{} 盈利金额{}",leadTraderId,orderId,instId,pnl);
        //更新跟单信息
        UserFollowTrader userFollowTrader = userFollowTraderRepository.findByUserIdAndLeadTraderId(userId,leadTraderId).orElse(null);
        if(userFollowTrader!=null){
            userFollowTrader.setFollowCount(userFollowTrader.getFollowCount()+1);
            // 如果盈利 > 0，则胜利次数 +1
            if (pnl.compareTo(BigDecimal.ZERO) > 0) {
                userFollowTrader.setWinCount(userFollowTrader.getWinCount() + 1);
            }
            // 累计盈利
            userFollowTrader.setTotalPnl(userFollowTrader.getTotalPnl().add(pnl));
            userFollowTraderRepository.save(userFollowTrader);
        }
    }


    /**
     * 根据仓位对象判断买卖方向
     * @param pos 仓位对象
     * @return "buy" 或 "sell"
     */
    public static String getPositionDirection(LeadTraderCurrentPositionsResponseDTO.Position pos) {
        if (pos == null) return null;

        String posSide = pos.getPosSide(); // long / short / net
        String subPosStr =pos.getSubPos(); // 仓位数量，可能是字符串

        // 尝试转换为 BigDecimal
        BigDecimal subPos = BigDecimal.ZERO;
        try {
            if (subPosStr != null && !subPosStr.isEmpty()) {
                subPos = new BigDecimal(subPosStr);
            }
        } catch (Exception e) {
            return null;
        }

        switch (posSide) {
            case "long":
                return "buy"; // 开多
            case "short":
                return "sell"; // 开空
            case "net":
                if (subPos.compareTo(BigDecimal.ZERO) > 0) {
                    return "buy"; // 正数代表开多
                } else if (subPos.compareTo(BigDecimal.ZERO) < 0) {
                    return "sell"; // 负数代表开空
                } else {
                    return null; // 仓位为0
                }
            default:
                return null;
        }
    }
    public static long getTodayStartMillisUTC() {
        LocalDate today = LocalDate.now(ZoneId.of("America/New_York"));
        long epochMilli = today.atStartOfDay(ZoneId.of("America/New_York")).toInstant().toEpochMilli();
        log.info("美国凌晨时间 {}",new Date(epochMilli));
        return epochMilli;
    }

    public static long getCurrentMillisNY() {
        ZoneId zone = ZoneId.of("America/New_York");
        long epochMilli = Instant.now().atZone(zone).toInstant().toEpochMilli();
        log.info("纽约当前时间 (America/New_York now): {}", new Date(epochMilli));
        return epochMilli;
    }
    /**
     * 获取当前盈利和亏损，进行止损止盈
     */
    //@Scheduled(cron = "0/3 * * * * ?")
    public void getProfit() throws Exception {
        log.info("获取当前盈利和亏损，进行止损止盈 时间{}",new Date());
        //获取当前自己的仓位
        List<OkxPositionResponseDTO.PositionData> data = new ArrayList<>();
        OkxPositionResponseDTO accountPositions = null;
        try {
            accountPositions = OkxPerpClient.getAccountPositions();
            if(accountPositions!=null && accountPositions.getCode().equals("0")){
                data = accountPositions.getData();
            }
        } catch (Exception e) {
            log.info("获取仓位信息有错误");
        }
        //清理不存在的数据
        cleanMargin(data);
        List<UserCopyTrade> openOrder = userCopyTradeRepository.findByStatus("OPEN");
        Map<String, UserCopyTrade> collectInstId = openOrder.stream()
                .collect(Collectors.toMap(
                        UserCopyTrade::getInstId,
                        k -> k,
                        (k1, k2) -> k1.getOpenTime().after(k2.getOpenTime()) ? k1 : k2
                ));
        if(CollectionUtils.isEmpty(openOrder)){
            log.info("当前时间{}未发现正在运行的订单",TradeSignalUtil.formatTimestamp(new Date().getTime()));
            return;
        }
        for (OkxPositionResponseDTO.PositionData pos : data) {
            if (pos.getUplRatio() == null || pos.getPos() == null) continue;
            UserCopyTrade userCopyTrade = collectInstId.get(pos.getInstId());
            boolean result = shouldTrigger(pos, userCopyTrade.getInstId(),userCopyTrade.getOpenTime(),new BigDecimal(pos.getUplRatio()),pos.getPosSide(),collectInstId);
            if(result){
                closeCurrentOrder(pos,collectInstId);
            }
        }

    }

    public  boolean queryResultPosition(String uniCode,long opentime,String instId) throws Exception {
        //获取交易员历史账单
        List<CopyTradingSubPositionsHistoryResponse.SubPosition> traderHistory = OkxPerpClient.getTraderHistory(uniCode,"");
        Optional<CopyTradingSubPositionsHistoryResponse.SubPosition> latestSubPositionOpt =
                traderHistory.stream()
                        .filter(s -> s.getOpenTime() != null) // 过滤掉空值
                        .max(Comparator.comparingLong(s -> Long.parseLong(s.getOpenTime())));
        //筛选符合条件的订单
        if(!CollectionUtils.isEmpty(traderHistory)){
            if(latestSubPositionOpt.isPresent()){
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                sdf.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
                CopyTradingSubPositionsHistoryResponse.SubPosition subPosition = latestSubPositionOpt.get();
                log.info("交易员{} 最新历史数据时间开仓时间{}",uniCode,sdf.format(new Date(Long.parseLong(subPosition.getOpenTime()))));
            }

            List<CopyTradingSubPositionsHistoryResponse.SubPosition> collectFilter = traderHistory.stream()
                    .filter(k -> k.getInstId().equals(instId))
                    .filter(k -> {
                        long kOpenTime = Long.parseLong(k.getOpenTime()); // 如果 getOpenTime 返回的是字符串
                        return Math.abs(kOpenTime - opentime) <= 1000; // 相差 1 秒以内
                    })
                    .collect(Collectors.toList());
            if(!CollectionUtils.isEmpty(collectFilter)){
                log.info("查找到当前跟单者{} 平仓订单，立即平仓币种{} 开仓时间{}",uniCode,instId,new Date(opentime));
                return Boolean.TRUE;
            }
        }
        return Boolean.FALSE;
    }

    public void closeCurrentOrder(OkxPositionResponseDTO.PositionData pos,  Map<String, UserCopyTrade> collectInstId) throws Exception {
        // 超过亏损或盈利阈值
        //平仓
        String position = pos.getPos();
        String posSide = pos.getPosSide();
        String tdMode = pos.getMgnMode();
        double posSize = Double.parseDouble(position);

        if (posSize == 0) {
            return;
        }
        String side;
        side = posSize > 0 ? "sell" : "buy";
        posSize = Math.abs(posSize);
        //市价单平仓
        OkxOrderResponseDTO okxOrderResponseDTO = OkxPerpClient.placeCloseOrder(pos.getInstId(), tdMode, posSide, side, String.valueOf(posSize));
        log.info("亏损和盈利平仓信息：{}",okxOrderResponseDTO);
        log.info("币种{} 平仓结果{}",pos.getInstId(),"0".equals(okxOrderResponseDTO.getCode())?"成功":"失败");
        UserCopyTrade userCopyTradeBean = collectInstId.get(pos.getInstId());
        if(userCopyTradeBean!=null){
            userCopyTradeBean.setStatus("CLOSE");
            userCopyTradeBean.setUpdateTime(new Date());
            if(okxOrderResponseDTO.getCode().equals("0")){
                userCopyTradeBean.setCloseOrdId(okxOrderResponseDTO.getData().get(0).getOrdId());
                userCopyTradeBean.setCloseTime(new Date(Long.parseLong(okxOrderResponseDTO.getData().get(0).getTs())));
                //查询订单盈利
                userCopyTradeBean.setPnlRatio(OkxPerpClient.getOrderPnl(userCopyTradeBean.getInstId(),okxOrderResponseDTO.getData().get(0).getOrdId()));
            }
            userCopyTradeRepository.save(userCopyTradeBean);
        }
    }
    /**
     * 判断开仓时间与收益率是否满足条件
     *
     * @param openTime 开仓时间
     * @param uplRatio 当前收益率 (0.20 = 20%)
     * @return true = 满足条件, false = 不满足
     */
    public  boolean shouldTrigger(OkxPositionResponseDTO.PositionData pos ,String instId, Date openTime, BigDecimal uplRatio,String side,Map<String, UserCopyTrade> collectInstId) throws Exception {
        if (openTime == null || uplRatio == null) {
            return false;
        }
        long now = System.currentTimeMillis();
        long diffMillis = now - openTime.getTime();
        // 转换成小时，保留两位小数
        BigDecimal diffHours = BigDecimal.valueOf(diffMillis / (1000.0 * 60 * 60))
                .setScale(2, RoundingMode.HALF_UP);

        if (uplRatio.compareTo(BigDecimal.valueOf(0.075)) >= 0
                ) {
            log.info("币种：{} 距离开仓小时数：{} 开仓时间 {}，条件满足: 盈利 >= 7.5%  自动平仓",
                    instId, diffHours, TradeSignalUtil.formatTimestamp(openTime.getTime()));
            return true;
        }
        if (uplRatio.compareTo(BigDecimal.valueOf(-0.0375)) <= 0) {
            log.info("币种：{} 距离开仓小时数：{} 开仓时间 {} 条件满足: 亏损 >= -3.5% 自动平仓",
                    instId,diffHours, TradeSignalUtil.formatTimestamp(openTime.getTime()));
            return true;
        }
        log.info("币种：{} 距离开仓小时数：{}  开仓时间 {}，条件都不满足: 平仓 当前亏盈{}",
                instId, diffHours,TradeSignalUtil.formatTimestamp(openTime.getTime()),uplRatio.multiply(BigDecimal.valueOf(100)) + "%");
        return false;
    }



    /**
     * 清理已经平仓的订单
     */
    //@Scheduled(cron = "0 * * * * ?")
    public void cleanMargin(List<OkxPositionResponseDTO.PositionData> data) {
        if( data.isEmpty()){
            return;
        }
        List<UserCopyTrade> openOrder = userCopyTradeRepository.findByStatus("OPEN");
        log.info("开始清理跟单员已经平仓的仓位 {} 数量{}", new Date(),openOrder.size());
        //清理仓位不存在的纪录
        if(!CollectionUtils.isEmpty(data)){
            Map<String, OkxPositionResponseDTO.PositionData> collectData = data.stream().collect(Collectors.toMap(OkxPositionResponseDTO.PositionData::getInstId, k1 -> k1));
            for(UserCopyTrade userCopyTrade :openOrder){
               if(collectData.get(userCopyTrade.getInstId())==null){
                   userCopyTrade.setStatus("CLOSED");
                   userCopyTrade.setUpdateTime(new Date());
                   userCopyTradeRepository.save(userCopyTrade);
                   log.info("清理仓位不存在的订单纪录 唯一码{}  跟单subPosid订单号 {}",userCopyTrade.getTraderId(),userCopyTrade.getSubPosId());
               }
           }
        }

    }

    /**
     * 将 UserCopyTrade 转换为 PositionData 用于平仓
     */
    private OkxPositionResponseDTO.PositionData

    convertToPositionData(UserCopyTrade trade) {
        OkxPositionResponseDTO.PositionData positionData = new OkxPositionResponseDTO.PositionData();
        positionData.setInstId(trade.getInstId());
        positionData.setPos(String.valueOf(trade.getPos()));
        positionData.setPosSide(trade.getPosSide());
        positionData.setMgnMode(trade.getMgnMode());
        return positionData;
    }

    public void handleClose(OkxPositionResponseDTO.PositionData positionData, Long userId, String keyUniCode) throws Exception {
        String position = positionData.getPos();
        String posSide = positionData.getPosSide();
        String tdMode = positionData.getMgnMode();
        double posSize = Double.parseDouble(position);

        if (posSize == 0) {
            return;
        }
        String side = posSize > 0 ? "sell" : "buy";
        posSize = Math.abs(posSize);

        OkxOrderResponseDTO okxOrderResponseDTO = OkxPerpClient.placeCloseOrder(
                positionData.getInstId(), tdMode, posSide, side, String.valueOf(posSize)
        );
        saveOrUpdateWatchTrader(userId, keyUniCode, positionData.getInstId(), okxOrderResponseDTO.getData().get(0).getOrdId());
        log.info("平仓结果 okxOrderResponseDTO {}", okxOrderResponseDTO);
    }

    public static long getCurrentUsEastMillis() {
        // 当前系统时间 -> 美国东部时间
        ZonedDateTime usEastTime = ZonedDateTime.now(ZoneId.of("America/New_York"));
        // 转成 Instant 再取毫秒
        return usEastTime.toInstant().toEpochMilli();
    }

    public Boolean validateOrderExist(Long  openTime,String uniCode,String subOrdId,String instId){
        List<UserCopyTrade> openOrder = userCopyTradeRepository.findByTraderId(uniCode);
        if(!CollectionUtils.isEmpty(openOrder)){
            List<UserCopyTrade> collectSubOrdId = openOrder.stream().filter(k -> k.getSubPosId().equals(subOrdId)).collect(Collectors.toList());
            if(collectSubOrdId.size()>0){
                log.info("当前订单存在，不需要再次下单 uniCode：{} subOrdId：{} ",uniCode,subOrdId);
                return Boolean.TRUE;
            }
            List<UserCopyTrade> collectInstId = openOrder.stream().filter(k -> k.getInstId().equals(instId) && Objects.equals(k.getLeadTraderOpenTime().getTime(), openTime)).collect(Collectors.toList());
            if(collectInstId.size()>0){
                log.info("该跟单员uniCode：{} 当前在当日订单存在相同的instId{}订单，不需要重复下单  subOrdId：{} 跟单者开单时间{} ",uniCode,instId,subOrdId,new Date(openTime));
                return Boolean.TRUE;
            }
        }
        return Boolean.FALSE;
    }

    public static long getTodayStartMillis() {
        LocalDate today = LocalDate.now();
        long epochMilli = today.atStartOfDay(ZoneId.systemDefault())
                .toInstant()
                .toEpochMilli();
        log.info("获取系统当前的时间{}",TradeSignalUtil.formatTimestamp(epochMilli));
        return epochMilli;
    }
}
